iT邦幫忙

2025 iThome 鐵人賽

DAY 13
0

昨天我先把 Memory 的骨架定義好,今天則輪到最核心的「Brain」。
這一層是 Controller 的角色:它不自己思考、不自己記錄,而是負責把 Memory 與 Thinking Net 黏合起來,驅動整個 Agent 的有限狀態機(FSM)。

換句話說:
Memory 是資料真相,State 是行為模組,Brain 則是整個 orchestrator。


設計目標

  • Plug and play:Thinking Net 與 Memory Structure 都能以字串組態載入,或直接塞自定義實例。
  • Stateless State × Controller 驅動:所有狀態遷移由 Brain 掌控,避免隱性轉移。
  • Loop 限制:確保不會無限卡死,對照實驗也更可控。
  • 觀測性:每一次 step 都可以串流輸出,清楚看到 Agent 在想什麼做什麼。

程式骨架

class Brain:
    def __init__(
        self,
        thinking_net: str | dict[Enum, State],
        memory_struct: str | Memory,
        tools: list[BaseTool],
    ) -> None:
        # 初始化 net 與 memory
        self.net = self._create_net(thinking_net)
        self.memory = self._create_memory(memory_struct, tools)

        # 初始狀態:直接取 net 的第一個 state
        self.state = list(self.net.values())[0]
        self.loop_limit = 10

    async def _step(self) -> AsyncIterator[str]:
        # 進入狀態
        self.state.on_enter(self.memory)

        # 執行狀態,並串流輸出
        async for chunk in self.state.run(self.memory):
            yield chunk

        # 離開狀態
        self.state.on_exit(self.memory)

        # 決定下一個狀態
        next_state_enum = await self.state.next_state(self.memory)
        self.state = self.net[next_state_enum]

    async def answer(self, task: str) -> AsyncIterator[str]:
        # 先把目標設定到 memory
        await self.memory.set_goal(task)

        while not self.memory.done and self.loop_limit > 0:
            async for chunk in self._step():
                yield chunk
            self.loop_limit -= 1

    def _create_net(self, thinking_net: str | dict[Enum, State]) -> dict[Enum, State]:
        if isinstance(thinking_net, str):
            if thinking_net not in NET_MAP:
                raise ValueError(f"Unknown thinking net: {thinking_net}")
            return NET_MAP[thinking_net]
        return thinking_net

    def _create_memory(self, memory_struct: str | Memory, tools: list[BaseTool]) -> Memory:
        if isinstance(memory_struct, str):
            if memory_struct not in MEM_MAP:
                raise ValueError(f"Unknown memory struct: {memory_struct}")
            return MEM_MAP[memory_struct](tools=tools)
        return memory_struct

設計細節

  1. Thinking Net

    • 可以直接給一個字串(如 "react"),由 NET_MAP 載入預設的 FSM 定義。
    • 也可以自己傳一個 dict[Enum, State],完全自定義狀態與轉移。
  2. Memory

    • 一樣支援字串(如 "messages")對應到 MEM_MAP,或直接丟進現成的 Memory 實例。
    • 這樣就能自由替換,例如:TreeMemory 搭配 ReflexionNet
  3. State 生命週期

    • 每個 _step() 都會呼叫 on_enterrun(串流輸出)→ on_exit
    • 接著再透過 next_state() 決定轉移。
    • 轉移權完全在 Brain 手上,State 本身不會偷偷改 Memory 以外的東西。
  4. Loop Limit

    • 目前先硬編成 10,避免意外無窮迴圈。
    • 之後可以做成可設定參數,或支援自動 early stop。

串接 Memory × Net 的示例

例如:

brain = Brain(
    thinking_net="react",          # ReAct FSM
    memory_struct="messages",      # 最暴力的訊息列表 Memory
    tools=[SearchTool(), CalcTool()]
)

async for chunk in brain.answer("幫我查台北今天天氣並算出明天帶傘機率"):
    print(chunk)

這樣就能看到:Reasoning → Action → Reasoning → Answering 的完整過程。


下一步

到這裡,Agent-brain 的最小骨架已經有了

  • Memory 可以選不同結構
  • Thinking Net 可以選不同 FSM
  • Brain 負責 orchestrator

明天目標:把 BaseTool 完成,補齊 execute() 和序列化邏輯。
有了工具,agent-brain 才能真正「動起來」,跑完整個 ReAct 流程。


上一篇
Day 12: agent-brain 的 Memory
系列文
agent-brain: 從 0 開始打造一個 python package13
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言